package top.doe.spark_sql;

import org.apache.spark.sql.Encoder;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.expressions.Aggregator;


public class MyAvg2 extends Aggregator<Double, Agg, Double> {


    @Override
    public Agg zero() {
        return new Agg();
    }

    @Override
    public Agg reduce(Agg b, Double a) {
        b.count += 1;
        b.sum += a;
        return b;
    }

    @Override
    public Agg merge(Agg b1, Agg b2) {
        b1.count += b2.count;
        b1.sum += b2.sum;

        return b1;
    }

    @Override
    public Double finish(Agg reduction) {

        return reduction.sum/reduction.count;
    }

    @Override
    public Encoder<Agg> bufferEncoder() {

        return Encoders.bean(Agg.class);
    }

    @Override
    public Encoder<Double> outputEncoder() {
        return Encoders.DOUBLE();
    }


}
